/*
 *  Copyright 2020-2025 the original author or authors.
 *  You cannot use this file unless authorized by the author.
 */

package org.ipig.commons.helper;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.ipig.commons.conf.kafka.producer.KafkaProducerConf;
import org.ipig.commons.conf.mongo.MongoConf;
import org.ipig.commons.conf.redis.RedisConf;
import org.ipig.commons.conf.spike.SpikeConf;
import org.ipig.commons.exception.UnsupportException;
import org.ipig.constants.ConnectorCnst;
import org.ipig.data.as.impl.SpikeManager;
import org.ipig.data.mongo.impl.MongoManager;
import org.ipig.data.redis.RedisManager;
import org.ipig.messaging.kafka.producer.GenericKafkaProducer;
import org.ipig.model.ConnectorConf;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * ConnectorHelper
 *
 * @author <a href="mailto:comchnts@163.com">chinats</a>
 * @version $$Id$$
 * @since 1.0
 */
@Slf4j
public class ConnectorHelper {
    /**
     * kafkaProducerConnectorMap
     */
    private static volatile Map<String, GenericKafkaProducer> kafkaProducerConnectorMap = new ConcurrentHashMap<>(8);
    /**
     * mongoConnectorMap
     */
    private static volatile Map<String, MongoManager> mongoConnectorMap = new ConcurrentHashMap<>(8);
    /**
     * spikeConnectorMap
     */
    private static volatile Map<String, SpikeManager> spikeConnectorMap = new ConcurrentHashMap<>(8);
    /**
     * redisConnectorMap
     */
    private static volatile Map<String, RedisManager> redisConnectorMap = new ConcurrentHashMap<>(8);


    /**
     * 获取kafka producer连接器
     *
     * @param connectorConf ConnectorConf
     * @return GenericKafkaProducer
     */
    public static GenericKafkaProducer getKafkaProducerConnector(ConnectorConf connectorConf) {
        if (connectorConf == null) {
            return null;
        }
        GenericKafkaProducer connector = null;
        if (StringUtils.equalsIgnoreCase(connectorConf.getConnType(), ConnectorCnst.ConnType.KAFKA_PRODUCER.code)) {
            KafkaProducerConf kafkaProducerConf = GsonHelper.fromJson(StringHelper.valueOf(connectorConf.getConnJson()), KafkaProducerConf.class);
            String key = kafkaProducerConf.getBootstrapServers() + kafkaProducerConf.getTopicName();
            connector = kafkaProducerConnectorMap.get(key);
            if (connector == null) {
                synchronized (ConnectorHelper.class) {
                    connector = kafkaProducerConnectorMap.get(key);
                    if (connector == null) {
                        connector = new GenericKafkaProducer();
                        connector.setConf(kafkaProducerConf);
                        kafkaProducerConnectorMap.put(key, connector);
                    }
                }
            }
        } else {
//            log.warn("不支持的连接器类型" + connectorConf.getConnType());
            throw new UnsupportException("不支持的连接器类型" + connectorConf.getConnType());
        }
        return connector;
    }


    /**
     * 获取kafka producer连接器
     *
     * @param conf KafkaProducerConf
     * @return GenericKafkaProducer
     */
    public static GenericKafkaProducer getKafkaProducerConnector(KafkaProducerConf conf) {
        if (conf == null) {
            return null;
        }
        String key = conf.getBootstrapServers() + conf.getTopicName();
        GenericKafkaProducer connector = kafkaProducerConnectorMap.get(key);
        if (connector == null) {
            synchronized (ConnectorHelper.class) {
                connector = kafkaProducerConnectorMap.get(key);
                if (connector == null) {
                    connector = new GenericKafkaProducer();
                    connector.setConf(conf);
                    kafkaProducerConnectorMap.put(key, connector);
                }
            }
        }
        return connector;
    }

    /**
     * 获取spike连接器
     *
     * @param connectorConf ConnectorConf
     * @return MongoClientPool
     */
    public static SpikeManager getSpikeConnector(ConnectorConf connectorConf) {
        if (connectorConf == null) {
            return null;
        }
        SpikeManager connector = null;
        if (StringUtils.equalsIgnoreCase(connectorConf.getConnType(), ConnectorCnst.ConnType.SPIKE.code)) {
            SpikeConf conf = GsonHelper.fromJson(StringHelper.valueOf(connectorConf.getConnJson()), SpikeConf.class);
            String key = conf.getServers();
            connector = spikeConnectorMap.get(key);
            if (connector == null) {
                synchronized (ConnectorHelper.class) {
                    connector = spikeConnectorMap.get(key);
                    if (connector == null) {
                        connector = new SpikeManager();
                        connector.setConf(conf);
                        connector.getClient();
                        spikeConnectorMap.put(key, connector);
                    }
                }
            }
        } else {
            log.warn("不支持的连接器类型{}", connectorConf.toString());
//            throw new UnsupportException("不支持的连接器类型" + connectorConf.toString());
        }
        return connector;
    }


    /**
     * 获取spike连接器
     *
     * @param conf SpikeConf
     * @return SpikeManager
     */
    public static SpikeManager getSpikeConnector(SpikeConf conf) {
        if (conf == null) {
            return null;
        }
        String key = conf.getServers();
        SpikeManager connector = spikeConnectorMap.get(key);
        if (connector == null) {
            synchronized (ConnectorHelper.class) {
                connector = spikeConnectorMap.get(key);
                if (connector == null) {
                    connector = new SpikeManager();
                    connector.setConf(conf);
                    connector.getClient();
                    spikeConnectorMap.put(key, connector);
                }
            }
        }
        return connector;
    }

    /**
     * 获取mongo连接器
     *
     * @param connectorConf ConnectorConf
     * @return MongoClientPool
     */
    public static MongoManager getMongoConnector(ConnectorConf connectorConf) {
        if (connectorConf == null) {
            return null;
        }
        MongoManager connector = null;
        if (StringUtils.equalsIgnoreCase(connectorConf.getConnType(), ConnectorCnst.ConnType.MONGODB.code)) {
            MongoConf conf = GsonHelper.fromJson(StringHelper.valueOf(connectorConf.getConnJson()), MongoConf.class);
            String key = conf.getAddress() + conf.getDbName() + conf.getUserName();
            connector = mongoConnectorMap.get(key);
            if (connector == null) {
                synchronized (ConnectorHelper.class) {
                    connector = mongoConnectorMap.get(key);
                    if (connector == null) {
                        connector = new MongoManager();
                        connector.setConf(conf);
                        connector.init();
                        mongoConnectorMap.put(key, connector);
                    }
                }
            }
        } else {
//            log.warn("不支持的连接器类型" + connectorConf.toString());
            throw new UnsupportException("不支持的连接器类型" + connectorConf.toString());
        }
        return connector;
    }


    /**
     * 获取mongo连接器
     *
     * @param conf MongoConf
     * @return MongoClientPool
     */
    public static MongoManager getMongoConnector(MongoConf conf) {
        if (conf == null) {
            return null;
        }
        String key = conf.getAddress() + conf.getDbName() + conf.getUserName();
        MongoManager connector = mongoConnectorMap.get(key);
        if (connector == null) {
            synchronized (ConnectorHelper.class) {
                connector = mongoConnectorMap.get(key);
                if (connector == null) {
                    connector = new MongoManager();
                    connector.setConf(conf);
                    connector.init();
                    mongoConnectorMap.put(key, connector);
                }
            }
        }
        return connector;
    }


    /**
     * 获取redis连接器
     *
     * @param connectorConf ConnectorConf
     * @return RedisManager
     */
    public static RedisManager getRedisConnector(ConnectorConf connectorConf) {
        if (connectorConf == null) {
            return null;
        }
        RedisManager connector = null;
        if (StringUtils.equalsIgnoreCase(connectorConf.getConnType(), ConnectorCnst.ConnType.REDIS.code)) {
            RedisConf conf = GsonHelper.fromJson(StringHelper.valueOf(connectorConf.getConnJson()), RedisConf.class);
            String key = conf.getHost() + conf.getPort();
            connector = redisConnectorMap.get(key);
            if (connector == null) {
                synchronized (ConnectorHelper.class) {
                    connector = redisConnectorMap.get(key);
                    if (connector == null) {
                        connector = new RedisManager(conf);
                        redisConnectorMap.put(key, connector);
                    }
                }
            }
        } else {
//            log.warn("不支持的连接器类型" + connectorConf.toString());
            throw new UnsupportException("不支持的连接器类型" + connectorConf.toString());
        }
        return connector;
    }

    /**
     * 获取redis连接器
     *
     * @param conf RedisConf
     * @return RedisManager
     */
    public static RedisManager getRedisConnector(RedisConf conf) {
        if (conf == null) {
            return null;
        }
        String key = conf.getHost() + conf.getPort();
        RedisManager connector = redisConnectorMap.get(key);
        if (connector == null) {
            synchronized (ConnectorHelper.class) {
                connector = redisConnectorMap.get(key);
                if (connector == null) {
                    connector = new RedisManager(conf);
                    redisConnectorMap.put(key, connector);
                }
            }
        }
        return connector;
    }
}

